package cn.hty.mymsg.handler.receiver.kafka;

import cn.hty.mymsg.constans.MessageQueuePipeline;
import cn.hty.mymsg.handler.receiver.MessageReceiver;
import cn.hutool.core.text.StrPool;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
import org.springframework.stereotype.Service;
import cn.hty.mymsg.handler.uitls.GroupIdMappingUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Optional;

/**
 * 启动消费者
 *
 */
//@Configuration
@Service
@ConditionalOnProperty(name = "austin.mq.pipeline", havingValue = MessageQueuePipeline.KAFKA)
@Slf4j
public class ReceiverStart implements MessageReceiver {
    /**
     * receiver的消费方法常量
     */
    private static final String RECEIVER_METHOD_NAME = "Receiver.consumer";
    /**
     * 获取得到所有的groupId
     */
    private static final List<String> GROUP_IDS = GroupIdMappingUtils.getAllGroupIds();
    /**
     * 下标(用于迭代groupIds位置)
     */
    private static Integer index = 0;
    @Autowired
    private ApplicationContext context;
    @Resource
    private ConsumerFactory consumerFactory;

    @Bean
    public static KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer groupIdEnhancer(){
        return (attrs,element)->{
            if (element instanceof Method){
                String name = ((Method) element).getDeclaringClass().getSimpleName() + StrPool.DOT + ((Method) element).getName();
                if (RECEIVER_METHOD_NAME.equals(name)) {
                    attrs.put("groupId", GROUP_IDS.get(index++));
                }
            }
            return attrs;
        };
    }
    /**
     * 为每个渠道不同的消息类型 创建一个Receiver对象
     */
    @PostConstruct
    public void init(){
        for (int i = 0; i < GROUP_IDS.size(); i++) {
            context.getBean(Receiver.class);
        }
    }
    /**
     * 针对tag消息过滤
     * producer 将tag写进header里
     *
     * @return true 消息将会被丢弃
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory filterContainerFactory(@Value("${austin.business.tagId.key}") String tagIdKey,
                                                                          @Value("${austin.business.tagId.value}") String tagIdValue){
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        //设置消息消费工厂
        factory.setConsumerFactory(consumerFactory);
        //设置消息确认策略
        factory.setAckDiscarded(true);
        //设置过滤策略
        factory.setRecordFilterStrategy(new RecordFilterStrategy<String, String>() {
            @Override
            public boolean filter(ConsumerRecord<String, String> consumerRecord) {
                if (Optional.ofNullable(consumerRecord.value()).isPresent()){
                    //遍历头信息，拿到键和值进行对比，相同保留，不同过滤
                    for (Header header : consumerRecord.headers()) {
                        if (header.key().equals(tagIdKey)&&new String(header.value(), StandardCharsets.UTF_8).equals(tagIdValue)){
                            return false;
                        }
                    }
                }
                return true;
            }
        });
        return factory;
    }
}
